package com.yxc.producer.handler;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.stereotype.Component;

/**
 * 消息发送成功失败回调
 *
 * @author: YXC
 * @time: 2022/12/6 10:53
 **/
@Slf4j
@Component
public class KafkaSendResultHandler implements ProducerListener<Object, Object> {

    /**
     * 成功回调
     *
     * @author YXC
     * @date 2022/12/6 10:54
     * @param producerRecord:
     * @param recordMetadata:
     * @return void
    **/
    @Override
    public void onSuccess(ProducerRecord<Object, Object> producerRecord, RecordMetadata recordMetadata) {
        if (log.isDebugEnabled()) {
            log.debug("消息发送成功  topic: {}, message: {}", producerRecord.topic(), producerRecord.value());
        }
    }

    /**
     * 失败回调
     *
     * @author YXC
     * @date 2022/12/6 10:54
     * @param producerRecord:
     * @param recordMetadata:
     * @param exception:
     * @return void
    **/
    @Override
    public void onError(ProducerRecord<Object, Object> producerRecord, RecordMetadata recordMetadata, Exception exception) {
        log.error("消息发送失败  topic: {}, message: {}", producerRecord.topic(), producerRecord.toString());
        ProducerListener.super.onError(producerRecord, recordMetadata, exception);
    }
}
